package com.mio.hbase;

import java.io.IOException;
import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

public class HbaseUtil {
	public static Configuration conf = null;
	public static Connection conn = null;
	public static HBaseAdmin admin = null;
	public static final String TABLE_NAME = "userfriends";
	public static final String COL_FAMILY_NAME = "friends";

	public static void init() throws IOException {
		conf = HBaseConfiguration.create();
		conf.set("hbase.zookeeper.quorum", "linux169,linux168,linux170");
		conf.set("hbase.zookeeper.property.clientPort", "2181");
		conf.set("hbase.client.start.log.errors.counter", "1");
		conf.set("hbase.client.retries.number", "1");
		conn = ConnectionFactory.createConnection(conf);
	}

	public static void destroy() {
		if (admin != null) {
			try {
				admin.close();
//				admin = null;
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		if (conn != null) {
			try {
				conn.close();
//				conn = null;
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	public static void createTable() throws IOException {
		if (admin == null) {
			HbaseUtil.init();
		}
		admin = (HBaseAdmin) conn.getAdmin();
		// 创建表描述器
		HTableDescriptor uf = new HTableDescriptor(TableName.valueOf(TABLE_NAME));
		// 设置列描述器
		uf.addFamily(new HColumnDescriptor(COL_FAMILY_NAME));
		// 创建
		admin.createTable(uf);
		System.out.println(TABLE_NAME + " create success!!");
	}

	public static void putData(String uid, List<String> friends) throws IOException {
		if (conn == null) {
			HbaseUtil.init();
		}
		// 获取一个表对象
		Table t = conn.getTable(TableName.valueOf(TABLE_NAME));
		// 设定rowkey
		Put put = new Put(Bytes.toBytes(uid));
		// 列族，列名 好友uid，value：好友uid
		for (String fuid : friends) {
			System.out.println("add frinds:" + fuid+"for "+uid);
			put.addColumn(Bytes.toBytes(COL_FAMILY_NAME), Bytes.toBytes(fuid), Bytes.toBytes(fuid));
		}
		t.put(put);
		// 关闭table对象
		t.close();
		System.out.println("插⼊成功!!");
	}
	
	public static void putData(String uid, String fuid) throws IOException {
		if (conn == null) {
			HbaseUtil.init();
		}
		// 获取一个表对象
		Table worker = conn.getTable(TableName.valueOf(TABLE_NAME));
		// 设定rowkey
		Put put = new Put(Bytes.toBytes(uid));
		// 列族，列名 好友uid，value：好友uid
		System.out.println("add frinds:" + fuid+"for "+uid);
		put.addColumn(Bytes.toBytes(COL_FAMILY_NAME), Bytes.toBytes(fuid), Bytes.toBytes(fuid));
		worker.put(put);
		// 关闭table对象
		worker.close();
		System.out.println("插⼊成功!!");
	}

	// 删除⼀一条数据

	public static void deleteData(String kuid, String fuid) throws IOException {
		// 需要获取⼀一个table对象
		final Table worker = conn.getTable(TableName.valueOf(TABLE_NAME));
		// 准备delete对象
		final Delete delete = new Delete(Bytes.toBytes(kuid));
		delete.addColumn(Bytes.toBytes(COL_FAMILY_NAME), Bytes.toBytes(fuid));
		
		// 执⾏删除
		worker.delete(delete);
		// 关闭table对象
		worker.close();
		System.out.println("删除数据成功!!");
		if(HbaseUtil.isColExist(fuid, kuid)) {
			HbaseUtil.deleteData(fuid, kuid);
		}
		
//		// 首先查看是否对于friends列族进行了删除
//				List<Cell> cells = delete.getFamilyCellMap().get(Bytes.toBytes(COL_FAMILY_NAME));
//				if (cells == null || cells.size() == 0) {
//					System.out.println("NO F famally operation ,just do it.");
//					return;
//				}
//
//				for (Cell cell : cells) {
////					HbaseUtil.deleteData(cell.getQualifierArray(), cell.getRowArray());
//					System.out.println("delete friends ");
//					HbaseUtil.deleteData(cell.getQualifier(), cell.getRow());
//				}
	}
	
	public static void deleteData(byte[] kuid, byte[] fuid) throws IOException {
		// 需要获取⼀一个table对象
		final Table worker = conn.getTable(TableName.valueOf(TABLE_NAME));
		// 准备delete对象
		final Delete delete = new Delete(kuid);
		delete.addColumn(Bytes.toBytes(COL_FAMILY_NAME), fuid);
		// 执⾏删除
		worker.delete(delete);
		// 关闭table对象
		worker.close();
		System.out.println("删除数据成功2!!");
	}

	public static void getDataByCF(String kuid) throws IOException {
		// 获取表对象
		Table uf =  conn.getTable(TableName.valueOf(TABLE_NAME));
		// 创建查询的get对象
		Get get = new Get(Bytes.toBytes(kuid));
		// 指定列族信息
//		get.addColumn(Bytes.toBytes(COL_FAMILY_NAME), Bytes.toBytes("sex"));
		get.addFamily(Bytes.toBytes(COL_FAMILY_NAME));
		// 执⾏查询
		Result res = uf.get(get);
		Cell[] cells = res.rawCells();
		// 获取改⾏的所有cell对象
		for (Cell cell : cells) {
			// 通过cell获取rowkey,cf,column,value
			String cf = Bytes.toString(CellUtil.cloneFamily(cell));
			String column = Bytes.toString(CellUtil.cloneQualifier(cell));
			String value = Bytes.toString(CellUtil.cloneValue(cell));
			String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
			System.out.println(rowkey + "----" + cf + "---" + column + "---" + value);
		}
		uf.close();
	}
	
	public static boolean isColExist(byte[] kuid,byte[] fuid) throws IOException {
		Table uf =  conn.getTable(TableName.valueOf(TABLE_NAME));
		//获取行键
        Get get2 = new Get(kuid);
        //获取列族
//        get2.addFamily(Bytes.toBytes(COL_FAMILY_NAME));
        get2.addColumn(Bytes.toBytes(COL_FAMILY_NAME), fuid);
        //设置校验是否存在 ==>Get 2 Exists: true
        get2.setCheckExistenceOnly(true);
        //获取数据集 ==>Get 2 Size: 0
        Result result2 = uf.get(get2);
 
        System.out.println("Get 2 Exists: " + result2.getExists());
        return result2.getExists();
	}
	
	public static boolean isColExist(String kuid,String fuid) throws IOException {
		Table uf =  conn.getTable(TableName.valueOf(TABLE_NAME));
		//获取行键
        Get get2 = new Get(Bytes.toBytes(kuid));
        //获取列族
        get2.addColumn(Bytes.toBytes(COL_FAMILY_NAME), Bytes.toBytes(fuid));
//        get2.addFamily(Bytes.toBytes(COL_FAMILY_NAME));
        //设置校验是否存在 ==>Get 2 Exists: true
        get2.setCheckExistenceOnly(true);
        //获取数据集 ==>Get 2 Size: 0
        Result result2 = uf.get(get2);
 
        System.out.println("Get 2 Exists: " + result2.getExists());
        return result2.getExists();
	}
}
